package com.etlmaster.executor.agent;

import java.util.concurrent.TimeUnit;

import com.etlmaster.executor.scheduler.Proc;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.scheduler.ScheduleStatus;
import com.etlmaster.executor.scheduler.utils.CallResetService;
import com.etlmaster.executor.scheduler.utils.SubmitResult;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.MultiProcessHelper;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class ZkAgent extends Agents {

	public ZkAgent(String id,String host, int maxThreadNum) {
		super(id,host, maxThreadNum);
		// TODO Auto-generated constructor stub
	}

	@Override
	public SubmitResult run(Proc proc) {
		SubmitResult result = new SubmitResult();
		if (this.getStatus() != ScheduleStatus.agentInvalid) {

			String url = "http://" + getHost() + "/etlmaster/startjob?jobjson="
					+ proc.getRunParam();
			CallResetService entity = new CallResetService(url, this, proc,
					template);
			getRunningProcs().add(proc);
			result.setResult(MultiProcessHelper.executor.submit(entity));
		}else{
			LogWriter.addLog("WARN","agent状态无效！不能接收新任务");
		}
		return result;

	}

	@Override
	public void stop() {
		ScheduleServer.getLock().lock();
		while (this.getRunningProcs().size() != 0) {
			try {
				LogWriter.addLog("WARN","{} 还有{}个任务在执行！",this.getId(),this.getRunningProcs().size()+"");
				ScheduleServer.getAgentNotEmpty().await(1,
						TimeUnit.MINUTES);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		String url = "http://" + getHost() + "/etlmaster/stopAgent";
		CallResetService entity = new CallResetService(url, this, null,
				template);
		MultiProcessHelper.executor.submit(entity);
		ScheduleServer.getLock().unlock();

	}

}
